package com.yesun.task.pipeline.handler.impl;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.yesun.core.util.FileUtils;
import com.yesun.server.mdp.service.client.config.HttpConfig;
import com.yesun.server.mdp.service.client.impl.DefaultClient;
import com.yesun.server.mdp.service.client.impl.http.HttpRequestContext;
import com.yesun.server.mdp.service.client.impl.http.HttpResponseContext;
import com.yesun.task.pipeline.common.Constant;
import com.yesun.task.pipeline.common.StockCode;
import com.yesun.task.pipeline.domain.dto.JobDto;
import com.yesun.task.pipeline.domain.dto.TaskDto;
import com.yesun.task.pipeline.domain.po.NewsPO;
import com.yesun.task.pipeline.handler.Executor;
import com.yesun.task.pipeline.handler.Handler;
import com.yesun.task.pipeline.handler.Loader;
import com.yesun.task.pipeline.service.impl.ClickhouseServiceImpl;
import com.yesun.task.pipeline.sink.ClickhouseSink;
import com.yesun.task.pipeline.worker.ParallelWorker;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.Header;
import org.apache.http.HttpEntity;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.BlockingQueue;

//东方财富新闻采集
@Slf4j
public class DfNewsHandler extends Handler {

    static DefaultClient client = new DefaultClient();
    static SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
    static SimpleDateFormat sdfDf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    static SimpleDateFormat shortSdf = new SimpleDateFormat("yyyyMMdd");
    static String TODAY = shortSdf.format(Calendar.getInstance().getTime());
    static Loader loader;

    public DfNewsHandler(JobDto jobDto) {
        super(jobDto);
    }

    @Override
    public Loader getLoader() {
        if(loader == null) {
            //实例化
            loader = new Loader() {
                @Override
                public void load(BlockingQueue queue) {
                    log.info("加载东方财富-资讯精华采集任务...");
                    //资讯精华，https://np-listapi.eastmoney.com/comm/web/getNewsByColumns?client=web&biz=web_news_col&column=345&order=1&needInteractData=0&page_index=1&page_size=20&req_trace=1694223062750&fields=code,showTime,title,mediaName,summary,image,url,uniqueUrl,Np_dst&types=1,20&callback=jQuery183017410379231098871_1694223062591&_=1694223062750
                    TaskDto taskDto = new TaskDto();
                    taskDto.setJobName(jobDto.getJobName());
                    taskDto.setName("东方财富-资讯精华采集任务");
                    taskDto.setTag(Constant.JOB_TAG_DF_IMPORTANT_NEWS);
                    Map<String,String> map = new HashMap<>();
                    map.put("source", "东方财富网");
                    map.put("category", "资讯精华");
                    map.put("url", "https://np-listapi.eastmoney.com/comm/web/getNewsByColumns?client=web&biz=web_news_col&column=345&order=1&needInteractData=0&page_index=1&page_size=50&req_trace="+System.currentTimeMillis()+"&fields=code,showTime,title,mediaName,summary,image,url,uniqueUrl,Np_dst&types=1,20");
                    taskDto.setMap(map);



                    //加入MAP_TASK
                    ParallelWorker.MAP_TASK.put(taskDto.getName(), taskDto.getJobName());


                    queue.offer(taskDto);
                    log.info("添加子任务：" + taskDto.getName());
                }
            };
        }
        return loader;
    }

    @Override
    public Executor getExecutor() {
        return new Executor() {
            @Override
            public void execute(TaskDto taskDto) {
                //执行task的采集
                log.info("execute：" + taskDto.getName());
                long start = System.currentTimeMillis();

                ClickhouseServiceImpl clickhouseServiceImpl = (ClickhouseServiceImpl) com.yesun.task.pipeline.common.SpringUtil.APPLICATION_CONTEXT.getBean("clickhouseServiceImpl");
                //从数据库读取当前分类最新记录
                String timestamp = "";
                List<Map<String, Object>> list = clickhouseServiceImpl.query("SELECT publish_at, topic, category, url FROM ods_stock_news WHERE dt>'20230901' AND category='"+taskDto.getMap().get("category")+"' ORDER BY publish_at DESC limit 1");
                if(list != null &&  list.size()>0) {
                    Map<String,Object> map = list.get(0);
                    timestamp = map.get("publish_at").toString();
                    log.info("timestamp:" + timestamp);
                }

                HttpConfig config = new HttpConfig(taskDto.getMap().get("url"), HttpConfig.HTTP_METHOD_GET);
                config.setEncoding("UTF-8");
                HttpRequestContext request = new HttpRequestContext(config);

                try {
                    HttpResponseContext response = (HttpResponseContext) client.execute(request);
                    String content = response.getContentAsString();
                    content = content.substring(content.indexOf("{"), content.lastIndexOf("}")+1);
                    JSONObject objRoot = JSONObject.parseObject(content);
                    if(objRoot != null) {
                        JSONObject objData = objRoot.getJSONObject("data");
                        if(objData != null) {
                            JSONArray jsonArray = objData.getJSONArray("list");
                            if (jsonArray != null) {
                                //只需要处理时间戳之后的
                                if(StringUtils.isNotBlank(timestamp)){
                                    for(int i=0;i<jsonArray.size();i++) {
                                        /**
                                         {"summary":"近日，中国证监会就活跃资本市场、提振投资者信心召开三场座谈会，分别听取专家学者和投资者意见建议，并就市场关切的热点问题做了交流探讨。证监会党委书记、主席易会满，党委委员、副主席李超以及证监会有关部门主要负责同志分别主持会议。来看会议关键内容：1、完善资本市场法治供给，加大资本市场防假打假力度，依法严惩欺诈发行、虚假披露、违规减持等违法违规行为。",
                                         "image":"https://np-newspic.dfcfw.com/download/D24723823541707120755_w210h154.jpg",
                                         "code":"202309092842529058",
                                         "np_dst":"CMS",
                                         "showTime":"2023-09-09 07:37:36",
                                         "uniqueUrl":"http://finance.eastmoney.com/a/202309092842529058.html",
                                         "title":"证监会连开三场座谈会！信息量巨大！将研究出台更多务实、管用的政策举措",
                                         "mediaName":"证券时报",
                                         "url":"http://finance.eastmoney.com/news/1344,202309092842529058.html"}
                                        **/
                                        JSONObject obj = (JSONObject) jsonArray.get(i);
//                                        log.info("showTime:" + obj.get("showTime").toString());
//                                        log.info("timestamp:" + timestamp);
                                        Date dtNew = sdfDf.parse(obj.get("showTime").toString());
                                        Date dtTimestamp = sdfDf.parse(timestamp);
                                        if(dtNew.compareTo(dtTimestamp)>0) {
                                            //表示，新的新闻
                                            //写入到缓存队列，等待批量写入到ck
                                            NewsPO newsPO = new NewsPO();
                                            newsPO.setDt(TODAY);
                                            newsPO.setCategory(taskDto.getMap().get("category").toString());
                                            newsPO.setTopic(obj.get("title").toString());
                                            newsPO.setSummary(obj.get("summary").toString());
                                            newsPO.setPublishAt(obj.get("showTime").toString());
                                            newsPO.setUrl(obj.get("url").toString());
                                            newsPO.setSource(taskDto.getMap().get("source").toString());
                                            ClickhouseSink.QUEUE_NEWS.offer(newsPO);
                                        } else {
//                                            log.info(obj.get("title").toString() + "，已存在！！！！！");
                                        }
                                    }
                                } else {
                                    for(int i=0;i<jsonArray.size();i++) {
                                        JSONObject obj = (JSONObject) jsonArray.get(i);
                                        //写入到缓存队列，等待批量写入到ck
                                        NewsPO newsPO = new NewsPO();
                                        newsPO.setDt(TODAY);
                                        newsPO.setCategory(taskDto.getMap().get("category").toString());
                                        newsPO.setTopic(obj.get("title").toString());
                                        newsPO.setSummary(obj.get("summary").toString());
                                        newsPO.setPublishAt(obj.get("showTime").toString());
                                        newsPO.setUrl(obj.get("url").toString());
                                        newsPO.setSource(taskDto.getMap().get("source").toString());
                                        ClickhouseSink.QUEUE_NEWS.offer(newsPO);
                                    }
                                }
                            }
                        }
                    }
                }catch (Exception e) {
                    e.printStackTrace();
                }

            }
        };
    }
}
